package org.eking.bigdata.kafka;

import java.util.Properties;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KStreamBuilder;
import org.apache.kafka.streams.processor.TopologyBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class kafkaMain {
	static final Logger logger = LoggerFactory.getLogger(kafkaMain.class);

	@SuppressWarnings("deprecation")
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		Properties settings = new Properties();
		// Set a few key parameters
		settings.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-first-streams-application");
		settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "10.71.200.109:9092");
		settings.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
		settings.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
		settings.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, "10.71.200.109:2182");
		
//		KStreamBuilder builder = new KStreamBuilder();
//		KStream<String, String> source1 = builder.stream("streams");
		TopologyBuilder builder = new TopologyBuilder();
		logger.info("begin");
		
		KafkaStreams streams = new KafkaStreams(builder, settings);
		builder.addSource("SOURCE", "streams").addProcessor("PROCESS1", () -> new kafkastream(), "SOURCE")
		.addSink("SINK1", "test", "PROCESS1");
		streams.start();
	}

}
